package com.feeyo.groundservice.vehiclemonitorserver;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.esri.core.geometry.MapGeometry;
import com.esri.core.geometry.OperatorCrosses;
import com.esri.core.geometry.OperatorWithin;
import com.esri.core.geometry.Point;
import com.esri.core.geometry.Polyline;
import com.esri.core.geometry.SpatialReference;
import com.feeyo.groundservice.base.rabbitmq.RmqSync;
import com.feeyo.groundservice.base.rdb.PostgresDb;
import com.feeyo.groundservice.base.redis.Redis;
import com.feeyo.groundservice.base.util.GlobalConstant;
import com.feeyo.groundservice.base.util.PropertiesUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import redis.clients.jedis.ShardedJedis;

public class MonitorService extends AbstractVerticle {
	
	Logger logger = LoggerFactory.getLogger(MonitorService.class);
	SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	//ShardedJedis jedis;
	public static final int coord_cache_threshold = 3; //坐标数量缓存个数阈值
	public static final int distance_threshold = 5;	   //两个连续坐标距离阈值（跑飞判断,单位米）
	public static final int time_interval_threshold = 20 * 1000;  //多个坐标回传时间的时间间隔阈值
	static Map<String,Long> devOffset = new ConcurrentHashMap<String,Long>();
	
	String exchange = PropertiesUtils.getProperty("rabbitmq.exchange");
	
	static List<String> whiteVehicles = new ArrayList<String>();
	Channel mqChannel = null;
	Channel produceChannel = null;
	RmqSync rmqSync = new RmqSync();
	ShardedJedis jedis = null;
	
	@Override
	public void start(Future<Void> startFuture) throws Exception {
		
		ScheduledExecutorService timeThreadPool = Executors.newScheduledThreadPool(1);//设备时间校准线程
		timeThreadPool.scheduleAtFixedRate(new Runnable() {
			@Override
			public void run() {
				devOffset.clear();
				System.out.println("clear-->>>>>>>>------>>>>>>>>");
			}
		}, 10*60, 10*60,TimeUnit.SECONDS);
		
		jedis = Redis.pool.getResource();
		
		mqChannel = rmqSync.getConnection().createChannel();
		produceChannel = rmqSync.getConnection().createChannel();
		mqChannel.basicConsume(PropertiesUtils.getProperty("rabbitmq.queue"),true, new DefaultConsumer(mqChannel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                try{
                    String message = new String(body, "UTF-8");
                    //logger.info("data tag->"+envelope.getRoutingKey());
                    //logger.info("data received->"+message);
                    if(message.startsWith("[")){
                    	JsonArray jos = new JsonArray(message);
                    	for(int i=0;i<jos.size();i++)
                    		consume(envelope.getRoutingKey(),jos.getJsonObject(i).toString());
                    }else
                    	consume(envelope.getRoutingKey(),message);
                    AlarmService.count++;
//                    JsonObject jo = new JsonObject(message);
                    	
                    //mqChannel.basicAck(envelope.getDeliveryTag(), false);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        });
		
        //vertx.eventBus().consumer("MONITOR", ar->{});
	}
	
	/**
	 * 
	 */
	public void ifCrossing(ShardedJedis jedis,Vehicle vehicle,JsonObject curCoord,List<Crossing> lines,String comp,String dept,String vehType){
		Double curLat = curCoord.getDouble("lat");
		Double curLon = curCoord.getDouble("lon");
		Double curSpeed = curCoord.getDouble("spd");
		List<String> historyCoordList = jedis.lrange(GlobalConstant.VEHICLE_HISTORY_COORDS + vehicle.getId(), 0,coord_cache_threshold-1);

		for(Crossing line:lines){
			List<CrossingPower> powers = line.getPowers();
			boolean flag = false;
			for(CrossingPower power : powers){
				if (vehicle.getDepartmentId() == null)
					return;
				if (vehicle.getTypeId() == null)
					return;
				if (power.getDepts().contains(vehicle.getDepartmentId()) && power.getTypes().contains(vehicle.getTypeId())){
					flag = true;
					break;
				}
			}
			if (!flag)
				continue;
			Polyline curLine = new Polyline();
			curLine.startPath(curLon, curLat);
			//当前车辆的类型和部门匹配当前车辆信息。
			JsonObject pre = new JsonObject(historyCoordList.get(historyCoordList.size()-1));
			curLine.lineTo(pre.getDouble("lon"),pre.getDouble("lat"));
			
			boolean isCross = OperatorCrosses.local().execute(line.getLine().getGeometry(), curLine, SpatialReference.create("4326"), null);
			if (isCross)
				log(jedis,vehicle.getId(),curCoord.getLong("time"),VehicleStatus.CROSS.getKey(),vehicle.getPosowner(),new JsonObject(){{put("limitArea", line.getId());}},
        				line.getId(),curSpeed,comp,dept,vehType,7);
		}
		
	}
	
	/**
	 * 
	 * <p>Description: </p>	
	 * @param vehicle 设备信息
	 * @param curCoord 当前坐标
	 * @param limitAreas 限制区域列表
	 */
	public void ifInArea(ShardedJedis jedis,Vehicle vehicle,JsonObject curCoord,List<Map<String, Object>> limitAreas,String comp,String dept,String vehType){
		Double curSpeed = curCoord.getDouble("spd");
	
        //获取设备对应的上次状态（是否超速，所在报警区域）
		String preAreaId = null;
		try{
			preAreaId = jedis.hget(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "limitArea");//java.lang.Long cannot be cast to [B
		}catch(ClassCastException e){
//			logger.info("cast error 1:"+vehicle.getId());
			e.printStackTrace();
		}
        Boolean preIsOverspeed = Boolean.parseBoolean(jedis.hget(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "overspeed"));
        
        //preAreaId如果删除了。。

		//此处暂用经纬度坐标
        //Point curPoint = new Point(curLon, curLat);
        String curAreaID = "";
        Double curLimitSpeed = 0.0;
        List<String> historyCoordList = null;
        try{
        	historyCoordList = jedis.lrange(GlobalConstant.VEHICLE_HISTORY_COORDS + vehicle.getId(), 0,coord_cache_threshold-1);
        }catch(ClassCastException e){
//        	logger.info("cast error 2:"+vehicle.getId());
			e.printStackTrace();
        }
        
		//遍历所有限制区
		for(Map<String,Object> limitArea : limitAreas){
			//***
			MapGeometry area = null;
			boolean deleted = false;
			String depts = "";
			String types = "";
			try{
	            area = (MapGeometry) limitArea.get("area"); //获取限制区
	            curAreaID = limitArea.get("id").toString();			    //获取限制区ID
	            curLimitSpeed = (Double)limitArea.get("limit_speed");	//获取区域限速
	            depts = (String)limitArea.get("depts");
	            types = (String)limitArea.get("types");
	            deleted = (boolean)limitArea.get("deleted");
			}catch(Exception e){
				logger.info("convertException3");
				e.printStackTrace();
			}
            
            String finalAreaID = curAreaID;
            
            boolean isInArea = true;  //是否在区域内 
            boolean isInTime = true;  //是否在时间范围内
            boolean isOverSpeed=true;
            
            if (vehicle.getDepartmentId() == null)
				return;
			if (vehicle.getTypeId() == null)
				return;
			
			if(depts != null){
				if (!depts.contains(vehicle.getDepartmentId())){
					continue;
				}
			}
			if(types != null){
				if(!types.contains(vehicle.getTypeId()))
					continue;
			}
            
            //当前区域限速大于0，且当前速度小于区域限速，且车辆在报警状态,当前区域和车辆报警区域相同。  解除超速报警状态
        	if (curSpeed < curLimitSpeed && curLimitSpeed >0 && curAreaID.equalsIgnoreCase(preAreaId)){
        		String tempArea = preAreaId;
        		log(jedis,vehicle.getId(),curCoord.getLong("time"),VehicleStatus.OVERSPEED_AREA_OUT.getKey(), vehicle.getPosowner(), new JsonObject(){{put("limitArea", tempArea);}},
        				preAreaId,curSpeed,comp,dept,vehType,3);
                jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "limitArea", "");
                jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "overspeed", "false");
                break;
        	}
        		
            if(!deleted){
	            for (int i=0;i< historyCoordList.size();i++){
	            	JsonObject historyCoord = new JsonObject(historyCoordList.get(i));
	            	//时间范围比对
	            	if ((curCoord.getLong("time")*1000 - historyCoord.getLong("time")) > time_interval_threshold ){
	            		isInTime = false;
	            		break;
	            	}
	               	Point historyPoint = new Point(historyCoord.getDouble("lon"),historyCoord.getDouble("lat"));
	            	//超速比对
	            	if (historyCoord.containsKey("speed"))
	              	  isOverSpeed = historyCoord.getDouble("speed") >= curLimitSpeed ? true : false;
	              	else{
	              		if (curLimitSpeed > 0) //curLimitArea > 0表示类型为限速区，=0为禁入区
	              			break;//continue;
	              	}
	            	//是否进入区域比对
	            	isInArea = OperatorWithin.local().execute(historyPoint, area.getGeometry(), SpatialReference.create("4326"), null);
	            	if (!isInArea)
	            		break;
	            	
//	            	if(!isOverSpeed)
//	            		break;
	            	
	            	if (isInArea){
	            		if (curLimitSpeed == 0)//在区域内，且区域限速为0表示此区域是禁入区
	            			continue;
	            		else if((curLimitSpeed > 0) && !isOverSpeed){  //在限制区内未超速
	            			////如果上次轨迹点在当前区域内，则取消报警。
	            			isInArea = false;
	            			break;
	            		}
	//            		if ((curLimitSpeed > 0) && isOverSpeed)  //在限制区内并超速
	//            			isInArea = true;
	            	}
	            	
	            }//end for(historyCoordList)
	            
	            if (!isInTime)
	            	break;
	            
	          //连续多个坐标在限制区内并超速
	            if (isInArea){
	            	//上次限制区和本次限制区不同。表示出了又进了一个区域。如果相同，表示还在当前限制区，无需报警。
	            	if (!curAreaID.equalsIgnoreCase(preAreaId)){
	            		
	            		if(curSpeed == 0){
	            			for (int i=0;i< historyCoordList.size();i++){
	        	            	JsonObject historyCoord = new JsonObject(historyCoordList.get(i));
	        	            	if (historyCoord.containsKey("speed")){
	        	            		if (historyCoord.getDouble("speed") >0){
	        	            			curSpeed = historyCoord.getDouble("speed");
	        	            			break;
	        	            		}
	        	            	}
	            			}
	            		}
	            		
	            		//上次状态是：在另一个限制区内，先写入出限制区事件
	                    if (preAreaId != null && preAreaId.length() > 0) {
	                    	String tempArea = preAreaId;
	                    	if (preIsOverspeed)//驶出超速区
	                    		log(jedis,vehicle.getId(),curCoord.getLong("time"), VehicleStatus.OVERSPEED_AREA_OUT.getKey(), vehicle.getPosowner(), new JsonObject(){{put("limitArea", tempArea);}},
	                    				preAreaId,curSpeed,comp,dept,vehType,3);
	                    	else			//驶出禁入区
	                    		log(jedis,vehicle.getId(),curCoord.getLong("time"),VehicleStatus.FORBID_AREA_OUT.getKey(), vehicle.getPosowner(), new JsonObject(){{put("limitArea", tempArea);}},
	                    				preAreaId,curSpeed,comp,dept,vehType,4);
	                        jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "limitArea", "");
	                        jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "overspeed", "false");
	                        preAreaId = "";
	                    }
	                    //logger.info("inArea:"+inAreaId+"||curAreaId:"+curAreaID);
	                    // 进入禁止区
	                    if (curLimitSpeed == 0) {
	                    	jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "overspeed", "false");//表示限制区类型
	                    	jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "eventid", String.valueOf(new Date().getTime()));//生成事件ID，更新驶出事件时所用
	                    	log(jedis,vehicle.getId(),curCoord.getLong("time"), VehicleStatus.FORBID_AREA_IN.getKey(), vehicle.getPosowner(), new JsonObject() {{
	                            put("limitArea", finalAreaID);
	                        }},finalAreaID,curSpeed,comp,dept,vehType,2);
	                    	
	                    } else { //进入超速区
	                    	double tempSpeed = curSpeed;
	                        JsonObject des = new JsonObject() {{
	                            put("limitArea", finalAreaID);
	                            put("speed", tempSpeed);
	                        }};
	                        jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "overspeed", "true");
	                        jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "eventid", String.valueOf(new Date().getTime()));
	                        
	                        //log(jedis,vehicle.getId(),curCoord.getLong("time"), VehicleStatus.OVERSPEED_AREA_IN.getKey(), vehicle.getPosowner(), 
//	                        		des,finalAreaID,curSpeed,comp,dept,vehType,1);
	                        JsonObject alarmCoord = new JsonObject(historyCoordList.get(coord_cache_threshold-1));
	                        log(jedis,vehicle.getId(),alarmCoord.getLong("time")/1000, VehicleStatus.OVERSPEED_AREA_IN.getKey(), vehicle.getPosowner(), 
	                        		des,finalAreaID,alarmCoord.getDouble("speed"),comp,dept,vehType,1);
	                    }
	                    jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "limitArea", curAreaID);
	                    
	            	}else{
	            		
	            	}
	                break;
	            }
            
            
            }
            
            //判定是否驶出限制区（连续多个坐标不在限制区，且限制区状态不为空）
            if (preAreaId == null || preAreaId.equals(""))
            	continue;
            if (!preAreaId.equals(curAreaID))
            	continue;
            
            isInArea = false;
            isInTime = true;
            for (int i=0;i< historyCoordList.size();i++){
            	JsonObject historyCoord = new JsonObject(historyCoordList.get(i));
            	if ((curCoord.getLong("time")*1000 - historyCoord.getLong("time")) > time_interval_threshold ){
            		isInTime = false;
            		break;
            	}
            	Point historyPoint = new Point(historyCoord.getDouble("lon"),historyCoord.getDouble("lat"));
//            		Polyline pl = new Polyline();
//            		pl.startPath(historyPoint);
//            		pl.lineTo(1, 1);
//            		OperatorCrosses.local().execute(pl, pl, null, null);
            	isInArea = OperatorWithin.local().execute(historyPoint, area.getGeometry(), SpatialReference.create("4326"), null);
            	if (isInArea)
            		break;
            
            }
            if (!isInTime)//超出时间阈值，无需关注
            	break;
            if (isInArea)
            	break; //以前为break;
            	
            //连续多个坐标不在限制区内
            if (!isInArea){
            	 //上次状态是：在另一个限制区内，写入出限制区事件
                 if (preAreaId != null && preAreaId.length() > 0) {
                	 	//logger.info("inArea:"+inAreaId+"||curAreaId:"+curAreaID);
                	 	String tempArea = preAreaId;
                    	if (preIsOverspeed)//驶出超速区
                    		log(jedis,vehicle.getId(),curCoord.getLong("time"),VehicleStatus.OVERSPEED_AREA_OUT.getKey(), vehicle.getPosowner(), new JsonObject(){{put("limitArea", tempArea);}},
                    				preAreaId,curSpeed,comp,dept,vehType,5);
                    	else			//驶出禁入区
                    		log(jedis,vehicle.getId(),curCoord.getLong("time"),VehicleStatus.FORBID_AREA_OUT.getKey(), vehicle.getPosowner(), new JsonObject(){{put("limitArea", tempArea);}},
                    				preAreaId,curSpeed,comp,dept,vehType,6);
                        jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "limitArea", "");
                        jedis.hset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehicle.getId(), "overspeed", "false");
                        preAreaId = "";
                 }
                 //break;
            }
            
            continue;
		}//end for(limitAreas)
		
		limitAreas = null;
		historyCoordList = null;
		
	}
	
	
	
    public  void log(ShardedJedis jedis,String vehicleId,long time, int eventType, String airport, JsonObject describe,String limitArea,Double speed,String comp,String dept,String vehType,int invokeLocation){
    	//int time =  curCo//Math.round(new Date().getTime() / 1000);
        //事件写入MQ  route_key:airport.comp.dept.vehType.vehicleId.EVENT
    	//logger.info("log->"+vehicleId + ":"+sdf.format(new Date(time*1000))+":"+ eventType+":"+limitArea+":invoke:"+invokeLocation);
    	
	        JsonObject body = new JsonObject()
	        		.put("_trigger_source",  new JsonObject(){{put("schema", "vehiclemonitor"); put("table", "vehicle"); put("pk", new JsonObject(){{put("id", vehicleId);}});}})
	        		.put("type", eventType)
	        		.put("description", describe)
	        		.put("time", time)
	                .put("speed",speed.intValue());
	        
	        try {
				produceChannel.basicPublish(airport.toUpperCase()+exchange,airport+"."+comp+"."+dept+"."+vehType+"."+vehicleId+".EVENT",null,body.toString().getBytes());
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
        
	        //事件写入关系数据库postgreSql
	        PostgresDb.postgreSQLClient.getConnection(getConn ->{
	            if (getConn.failed()){
	                System.out.println(getConn.cause());
	            }
	            if (getConn.succeeded()){
	        		String eventid = jedis.hget(GlobalConstant.VEHICLE_REALTIME_STATUS+vehicleId, "eventid");
	        		
	            	if ((eventType == VehicleStatus.OVERSPEED_AREA_IN.getKey())||(eventType == VehicleStatus.FORBID_AREA_IN.getKey())){
	            		logger.info("AREA_IN_EVENT1:"+vehicleId + ":"+sdf.format(new Date(time*1000))+":"+ eventType+":"+limitArea+":invoke:"+invokeLocation);
	                    getConn.result().updateWithParams("insert into vehiclemonitor.vehicleevent (_trigger_source,type,description, intime,vehicleid,maxspeed,area,eventid) "
	                    		+ " values (?::jsonb,?,?::jsonb,?,?::uuid,?,?::uuid,?)",
	                    	new JsonArray(){{
	                        add(new JsonObject(){{put("schema", "vehiclemonitor"); put("table", "vehicle"); put("pk", new JsonObject(){{put("id", vehicleId);}});}}.toString());
	                        add(eventType);
	                        add(describe.toString());
	                        add(time);
	                        add(vehicleId);
	                        add(speed.intValue());
	                        add(limitArea);
	                        add(Long.parseLong(eventid));
	                    	}},
	                    	r->{
	                    		getConn.result().close();
	                    });
	            	}else if ((eventType == VehicleStatus.OVERSPEED_AREA_OUT.getKey())||(eventType == VehicleStatus.FORBID_AREA_OUT.getKey())){
	            		logger.info("AREA_OUT_EVENT1:"+vehicleId + ":"+sdf.format(new Date(time*1000))+":"+ eventType+":"+limitArea+":invoke:"+invokeLocation);
	            		if (!(eventid == null)&&(!eventid.equals(""))){
		                    getConn.result().updateWithParams("update vehiclemonitor.vehicleevent SET outtime=? WHERE eventid=?",
		                    	new JsonArray(){{
		                        add(time);
		                        add(Long.parseLong(eventid));
		                    	}},
		                    	r->{
		                    		getConn.result().close();
		                    });
	            		}
	            	}else if (eventType == VehicleStatus.CROSS.getKey()){
	            		logger.info("CROSS_EVENT:"+vehicleId + ":"+sdf.format(new Date(time*1000))+":"+ eventType+":"+limitArea+":invoke:"+invokeLocation);
	                    getConn.result().updateWithParams("insert into vehiclemonitor.vehicleevent (_trigger_source,type,description, intime,vehicleid,maxspeed,area,eventid) "
	                    		+ " values (?::jsonb,?,?::jsonb,?,?::uuid,?,?::uuid,?)",
	                    	new JsonArray(){{
	                        add(new JsonObject(){{put("schema", "vehiclemonitor"); put("table", "vehicle"); put("pk", new JsonObject(){{put("id", vehicleId);}});}}.toString());
	                        add(eventType);
	                        add(describe.toString());
	                        add(time);
	                        add(vehicleId);
	                        add(speed.intValue());
	                        add(limitArea);
	                        add(Long.parseLong("0"));
	                    	}},
	                    	r->{
	                    		getConn.result().close();
	                    });
	            	}
	            	else{
	            		logger.info("AREA_EVENT_UNKNOWN:"+vehicleId + ":"+ eventType+":"+limitArea+":invoke:"+invokeLocation);
	            	}
	
	            }
	        });

    }
    
    public void consume(String routeKey,String messageData){
    	//***
        String[] keys = routeKey.split("[.]");
        if(keys.length < 7){  
            return;
        }
        String dataType = keys[5]; //pos(位置) state（状态）
        String dataRT = keys[6];  //RT实时 AD历史的
        String devid = keys[4].toUpperCase();
       
        keys = null;
        if(dataType.equalsIgnoreCase("POS")) {
                JsonObject curCoord = new JsonObject(messageData);
                //***
                Double lon = 0.0;
                Double lat = 0.0;
                Double speed = 0.0;
                Double dire = 0.0;
                double devTime = 0.0;   
                try{
                	lon = curCoord.getDouble("lon");
                	lat = curCoord.getDouble("lat");
                    speed = curCoord.getDouble("spd");;
                    dire = curCoord.getDouble("dir");
                    devTime = curCoord.getDouble("time");   
                }catch(Exception e){
                	logger.info("convertException2");
                	e.printStackTrace();
                }
                
                long serverTime = System.currentTimeMillis()/1000; 
                long interval = Math.round(devTime - serverTime);   
                
                if(lon == 0.0 || lat == 0.0)
                	return;
                
                if (interval > 5*60){	//设备时间校准
                	if (devOffset.get(devid) == null){
                		devOffset.put(devid, interval);
                		//logger.info("stamp->"+devid+"->"+interval);
                	}
                	devTime = devTime - devOffset.get(devid);
                	curCoord.put("time", devTime);
                }
                
                final Double time = devTime;
                Boolean located = curCoord.getBoolean("location");//是否定位成功
                if((located == null) || (located == false))
                	return;
                Vehicle vehicle = VehicleInfo.vehicleInfos.get(devid);                    
                
                if(vehicle == null){
                	return;
                }
                String vehId = vehicle.getId();
                JsonArray posJson = new JsonArray();
                posJson.add(vehId)
                		.add(lat)
                		.add(lon)
                		.add(speed)
                		.add(dire)
                		.add(-1)
                		.add(0)
                		.add(time*1000)
                		.add(dataRT);
                //经纬度，速度等入库influxdb
                vertx.eventBus().publish("tots",posJson);
                                    
                //非实时数据丢弃
                if(!dataRT.equalsIgnoreCase("RT") )
                    return;
                
                //从内存获取当前设备对应的机场，航空公司，部门信息，车辆类型
                String airport = vehicle.getPosowner();
                String comp = "";
                String dept = "";
                String vehType = "";
                
                Department.DepartmentObject departmentObject = Department.departments.get(vehicle.getDepartmentId());
                if(departmentObject != null){
                    dept = departmentObject.id;
                    Department.DepartmentObject compObject = Department.departments.get(departmentObject.parentId);
                    if(compObject != null){
                        comp = compObject.id;
                    }
                }
                
                VehicleType.VehicleTypeObject vehicleTypeObject = VehicleType.vehicleTypes.get(vehicle.getTypeId());
                if(vehicleTypeObject != null){
                    vehType = vehicleTypeObject.id;
                }
                
                Map<String,String> posMap = new HashMap<String, String>();
		                posMap.put("lat", lat.toString());
		                posMap.put("lon", lon.toString());
		                posMap.put("dire", dire.toString());
		                posMap.put("spd", speed.toString());
		                posMap.put("time", time.toString());
		                posMap.put("located", located.toString());
                try{
	                // 最新设备坐标写入REDIS 
	                jedis.hmset(GlobalConstant.VEHICLE_REALTIME_STATUS + vehId, posMap);// java.lang.Long cannot be cast to [B
                }catch(ClassCastException e){
                	//logger.info("cast error 3");
                	e.printStackTrace();
                }
                
                //最新设备坐标发布到MQ
                final String route_key = airport+"."+comp+"."+dept+"."+vehType+"."+dataType;
                String body = new JsonObject()
                		.put("id", vehId)
                		.put("lat", lat)
                		.put("lon", lon)
                		.put("dire", dire)
                		.put("spd", speed)
                		.put("time", time)
                		.put("located", located)
                		.put("number", vehicle.getNumber()).toString();
                
                try {
                	//produceChannel.exchangeDeclarePassive(airport.toUpperCase()+exchange);
					produceChannel.basicPublish(airport.toUpperCase()+exchange,route_key,null,body.toString().getBytes());
				} catch (Exception e) {
					// TODO Auto-generated catch block
					logger.info("exchange error1");
					try {
						produceChannel = rmqSync.getConnection().createChannel();
						produceChannel.exchangeDeclare(airport.toUpperCase()+exchange,"topic");
					} catch (IOException e1) {
						logger.info("exchange error2");
						e1.printStackTrace();
					}
					e.printStackTrace();
				} 
                
                if(speed>100)
                	return;
                
                JsonObject posCache = new JsonObject()
                		.put("lat", lat)
                		.put("lon",lon)
                		.put("speed",speed)
                		.put("time",time*1000);
                try{
	                //写入缓存，过去若干坐标点
	                jedis.lpush(GlobalConstant.VEHICLE_HISTORY_COORDS + vehId, posCache.toString()); //[B cannot be cast to java.lang.Long
                }catch(ClassCastException e){
                	//logger.info("cast error 3");
                	e.printStackTrace();
                }
                try{
	                jedis.ltrim(GlobalConstant.VEHICLE_HISTORY_COORDS + vehId, 0, coord_cache_threshold ); 
                }catch(ClassCastException e){
                	//logger.info("cast error 3");
                	e.printStackTrace();
                }
                
                //加入白名单功能 vehicle.getNumber()
                if (whiteVehicles.contains(vehicle.getNumber()))
                	return;
                
                //获取当前机场限制区列表
                List<Map<String, Object>> limitsCache = LimitAreaManager.limits.get(airport);
                if (limitsCache != null) 
                	ifInArea(jedis,vehicle, curCoord, limitsCache,comp,dept,vehType);
                
                List<Crossing> lines = CrossingManager.lines.get(airport);
                if (lines !=null)
                	ifCrossing(jedis,vehicle, curCoord, lines,comp,dept,vehType);
            
        }else{
            //System.out.println("loc msg");
        }
    }
	
	@Override
	public void stop(Future<Void> stopFuture) throws Exception {
		//jedis.close();
	}
	
	static {
		whiteVehicles.add("民航D6716");
		whiteVehicles.add("民航D6717");
		whiteVehicles.add("民航D4655");
		whiteVehicles.add("民航D6147");
		whiteVehicles.add("云AR883R");
		whiteVehicles.add("云AR899R");
		whiteVehicles.add("民航D3290");
		whiteVehicles.add("云A0D3W1");
		whiteVehicles.add("云A457FA");
		whiteVehicles.add("民航D6027");
		whiteVehicles.add("云AV375H");
		whiteVehicles.add("民航D5187");
		whiteVehicles.add("民航D5188");
		whiteVehicles.add("民航D4628");
		whiteVehicles.add("民航D4613");
		whiteVehicles.add("民航D6670");
		whiteVehicles.add("民航D6025");
		whiteVehicles.add("民航D4898");
		whiteVehicles.add("民航D7291");
		whiteVehicles.add("民航D7280");
		whiteVehicles.add("民航D7281");
		whiteVehicles.add("民航D6633");
		whiteVehicles.add("民航D6634");//云A457FA
		whiteVehicles.add("民航D6941");
		whiteVehicles.add("民航D6943");
		whiteVehicles.add("民航D5137");
		whiteVehicles.add("云A340JW");
		whiteVehicles.add("民航D4655");
		whiteVehicles.add("云AR885R");
		whiteVehicles.add("云A7F7Y0");
	}
	
	
}
